package com.example.controller;

import com.example.config.FlinkConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;

/**
 * Flink测试控制器
 */
@RestController
@RequestMapping("/api/flink")
@Slf4j
public class FlinkTestController {

    @Autowired
    private FlinkConfig flinkConfig;

    /**
     * 测试Flink连接
     */
    @PostMapping("/test-connection")
    public Map<String, Object> testFlinkConnection() {
        Map<String, Object> result = new HashMap<>();
        
        try {
            log.info("正在测试Flink连接到 {}:{}", 
                    flinkConfig.getRemote().getHost(), 
                    flinkConfig.getRemote().getPort());
            
            // 创建远程执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
                    flinkConfig.getRemote().getHost(),
                    flinkConfig.getRemote().getPort()
            );
            
            if (env != null) {
                result.put("status", "success");
                result.put("message", "Flink连接测试成功");
                result.put("flinkHost", flinkConfig.getRemote().getHost());
                result.put("flinkPort", flinkConfig.getRemote().getPort());
                log.info("Flink连接测试成功");
            } else {
                result.put("status", "error");
                result.put("message", "无法创建Flink执行环境");
            }
            
        } catch (Exception e) {
            log.error("Flink连接测试失败", e);
            result.put("status", "error");
            result.put("message", "Flink连接测试失败: " + e.getMessage());
            result.put("error", e.getClass().getSimpleName());
        }
        
        return result;
    }
} 